package com.ccc.mq.single.service;

import com.ccc.mq.single.send.SysCenterMQSend;
import com.ccc.test.common.constant.MQConstant;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * 中心级MQ服务器-Service
 *
 * @author superC
 * @version 1.0.0
 * @date 2022/6/30
 */
@Component
public class SysProducerService {

    protected static final Logger logger = LoggerFactory.getLogger(SysProducerService.class);

    @Autowired
    private SysCenterMQSend sysCenterMQSend;

    /**
     * 异步发送对象数据
     *
     * @param data  消息对象数据
     * @param topic 主题（formats: `topicName` / `topicName:tags`）
     * @date 2022/6/30 15:54
     * @author superC
     */
    public <T> void asyncSend(T data, String topic) {
        Message<T> message = MessageBuilder
                .withPayload(data)
                .build();

        logger.info("=======消息生产者========");
        logger.info("当前为【syncData】方法，发送topic为【{}】", topic);
        logger.info("===============");

        // 异步发送。
        sysCenterMQSend.asyncSend(topic, message);
    }

    /**
     * 异步发送对象数据
     *
     * @param data   消息对象数据
     * @param topic  主题（formats: `topicName` / `topicName:tags`）
     * @param dataId KEY-当前数据主键ID
     * @date 2022/6/30 15:54
     * @author superC
     */
    public <T> void asyncSend(T data, String topic, String dataId) {
        Message<T> message = MessageBuilder
                .withPayload(data)
                // 设置Key，用于消费端进行接口幂等性校验时使用。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
                .setHeader(RocketMQHeaders.KEYS, dataId)
                .build();

        logger.info("=======消息生产者========");
        logger.info("当前为【syncData】方法，发送topic为【{}】", topic);
        logger.info("===============");

        // 异步发送。
        sysCenterMQSend.asyncSend(topic, message);
    }

    /**
     * 异步发送对象数据——使用自定义Header属性过滤消息
     *
     * @param data       消息对象数据
     * @param topic      主题（formats: `topicName` / `topicName:tags`）
     * @param dataId     KEY-当前数据主键ID
     * @param methodType 自定义参数-操作类型（ADD、UPDATE、DELETE）
     * @date 2022/6/30 15:54
     * @author superC
     */
    public <T> void asyncSend(T data, String topic, String dataId, String methodType) {
        Message<T> message = MessageBuilder
                .withPayload(data)
                // 设置Key，用于消费端进行接口幂等性校验时使用。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
                .setHeader(RocketMQHeaders.KEYS, dataId)
                // 设置自定义参数，用于进行消息过滤，实现消息的分流处理。
                .setHeader(MQConstant.HEADER_METHOD, methodType)
                .build();

        logger.info("=======消息生产者========");
        logger.info("当前为【syncData2TagsAndHeader】方法，发送topic为【{}】，HEADER_METHOD类型为 {}", topic, methodType);
        logger.info("===============");

        // 异步发送。
        sysCenterMQSend.asyncSend(topic, message);
    }

    /**
     * 异步发送JSON数据
     *
     * @param dataJson 消息JSON数据
     * @param topic    主题（formats: `topicName` / `topicName:tags`）
     * @param dataId   KEY-当前数据主键ID
     * @date 2022/6/30 15:54
     * @author superC
     */
    public void asyncSendJson(String dataJson, String topic, String dataId) {
        Message<String> message = MessageBuilder
                .withPayload(dataJson)
                // 设置Key，用于消费端进行接口幂等性校验时使用。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
                .setHeader(RocketMQHeaders.KEYS, dataId)
                .build();

        logger.info("=======消息生产者========");
        logger.info("当前为【syncData】方法，发送topic为【{}】", topic);
        logger.info("===============");

        // 异步发送。
        sysCenterMQSend.asyncSend(topic, message);
    }

    /**
     * 异步发送JSON数据——使用自定义Header属性过滤消息
     *
     * @param dataJson   消息JSON数据
     * @param topic      主题（formats: `topicName` / `topicName:tags`）
     * @param dataId     KEY-当前数据主键ID
     * @param methodType 自定义参数-操作类型（ADD、UPDATE、DELETE）
     * @date 2022/6/30 15:54
     * @author superC
     */
    public void asyncSendJson(String dataJson, String topic, String dataId, String methodType) {
        Message<String> message = MessageBuilder
                .withPayload(dataJson)
                // 设置Key，用于消费端进行接口幂等性校验时使用。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
                .setHeader(RocketMQHeaders.KEYS, dataId)
                // 设置自定义参数，用于进行消息过滤，实现消息的分流处理。
                .setHeader(MQConstant.HEADER_METHOD, methodType)
                .build();

        logger.info("=======消息生产者========");
        logger.info("当前为【syncData2TagsAndHeader】方法，发送topic为【{}】，HEADER_METHOD类型为 {}", topic, methodType);
        logger.info("===============");

        // 异步发送。
        sysCenterMQSend.asyncSend(topic, message);
    }

    /**
     * 异步模式发送请求，并接收MQ的回复
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @date 2022/7/7 9:10
     * @author superC
     */
    public boolean syncSendOrderly(String topic, Object payload, String hashKey) {
       return sysCenterMQSend.syncSendOrderly(topic, payload, hashKey);
    }

    /**
     * 单向模式发送JSON数据
     *
     * @param dataJson   消息JSON数据
     * @param topic      主题（formats: `topicName` / `topicName:tags`）
     * @param dataId     KEY-当前数据主键ID
     * @param methodType 自定义参数-操作类型（ADD、UPDATE、DELETE）
     * @date 2022/7/7 9:48
     * @author superC
     */
    public void syncSend(String dataJson, String topic, String dataId, String methodType) {
        Message<String> msg = MessageBuilder
                .withPayload(dataJson)
                // 设置Key，用于消费端进行接口幂等性校验时使用。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
                .setHeader(RocketMQHeaders.KEYS, dataId)
                // 设置自定义参数，用于进行消息过滤，实现消息的分流处理。
                .setHeader(MQConstant.HEADER_METHOD, methodType)
                .build();
        sysCenterMQSend.convertAndSend(topic, msg);
    }

}
